package com.datalinkx.sse.config;

import ch.qos.logback.classic.spi.STEUtil;
import com.datalinkx.common.utils.ObjectUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 山顶洞人郑某
 * @className SseEmitterServer.java SSE 服务端服务
 * @date 2025 03 05
 */
@Slf4j
public class SseEmitterServer {
    /**
     * 当前连接数统计
     */
    private static final AtomicInteger count = new AtomicInteger(0);

    /**
     * 根据用户ID来获取当前对应的连接的SseEmitterServer对象，或者也可以放入Redis中
     */
    private static final Map<String,SseEmitter> sseEmitterServerMap = new ConcurrentHashMap<>();

    /**
     * 创建用户连接并返回SseEmitter对象
     * @param id 用户id
     * @return SseEmitter对象
     */
    public static SseEmitter connect(String id) {
        if (!ObjectUtils.isEmpty(sseEmitterServerMap.get(id))) {
            return sseEmitterServerMap.get(id);
        }
        SseEmitter sseEmitter = new SseEmitter(0L);
        //注册回调
        sseEmitter.onCompletion(() -> {
            log.info("[{}]连接已关闭！", id);
            removeUser(id);
        });
        sseEmitter.onError((e) -> {
            log.info("[{}]连接异常！", id);
            removeUser(id);
        });
        sseEmitter.onTimeout(() -> {
            log.info("[{}]连接超时！", id);
            removeUser(id);
        });
        sseEmitterServerMap.put(id, sseEmitter);
        count.incrementAndGet();
        return sseEmitter;
    }


    /**
     * 发送消息到客户端
     * @param id
     * @param message
     */
    public static void sendMessage(String id, String message) {
        try
        {
            SseEmitter sseEmitter = sseEmitterServerMap.get(id);
            if (sseEmitter == null) {
                sseEmitter = connect(id);
            }
            sseEmitter.send(message, MediaType.APPLICATION_JSON);
        }
        catch (IOException e) {
            log.error("[{}]发送消息失败！", id);
            removeUser(id);
        }
    }
    /**
     * 删除连接并减少连接数
     * @param id
     */
    public static void removeUser(String id) {
        //删除连接
        SseEmitter emitter = sseEmitterServerMap.get(id);
        if (emitter != null) {
            emitter.complete();
        }
        sseEmitterServerMap.remove(id);
        count.decrementAndGet();
    }

}
